package com.example.jdbcdemo.schedule.task;

import cn.hutool.core.collection.CollUtil;
import com.example.jdbcdemo.util.JdbcUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.sql.PreparedStatement;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;

@Service
@Slf4j
public class DataMigrationTaskService {

    @Resource(name = "resourceJdbcTemplate")
    private JdbcTemplate resourceJdbcTemplate;

    @Resource(name = "targetJdbcTemplate")
    private JdbcTemplate targetJdbcTemplate;

    @Value("${migrate.tables}")
    private List<String> migrateTableList;

    @Scheduled(fixedDelay = 1000*60*60*24)
    public void dataMigration() {
        long startTime = System.currentTimeMillis();

        String sql = "SELECT * FROM TABLE_NAME";
        for (String migrateTable : migrateTableList) {
            long t0 = System.currentTimeMillis();
            sql = sql.replace("TABLE_NAME", migrateTable);
            String finalSql = sql;

            // 截断表
            String truncateSql = "truncate table truncateTable";
            truncateSql = truncateSql.replace("truncateTable", migrateTable);
            targetJdbcTemplate.execute(truncateSql);

            List<CompletableFuture<Boolean>> futures = new ArrayList<>();

            List<Map<String, Object>> mapEntities = new ArrayList<>();
            resourceJdbcTemplate.query(con -> {
                con.setAutoCommit(false);
                PreparedStatement ps = con.prepareStatement(finalSql);
                ps.setFetchSize(20000);
                return ps;
            }, rs -> {
                do {
                    mapEntities.add(JdbcUtils.getResultMap(rs));
                    if (mapEntities.size() % 5000 == 0) {
                        List<Map<String, Object>> mapEntitieList = new ArrayList<>(mapEntities);
                        mapEntities.clear();
                        CompletableFuture<Boolean> f1 = CompletableFuture.supplyAsync(() -> {
                            log.info("插入一万条开始");
                            long t1 = System.currentTimeMillis();
                            JdbcUtils.batchInsertLarge(migrateTable, mapEntitieList, targetJdbcTemplate);
                            log.info("插入一万条耗时：{}", System.currentTimeMillis() - t1);
                            return Boolean.TRUE;
                        });
                        futures.add(f1);
                    }
                } while (rs.next());

                if (CollUtil.isNotEmpty(mapEntities)) {
                    CompletableFuture<Boolean> f2 = CompletableFuture.supplyAsync(() -> {
                        JdbcUtils.batchInsertLarge(migrateTable, mapEntities, targetJdbcTemplate);
                        return Boolean.TRUE;
                    });
                    futures.add(f2);
                    mapEntities.clear();
                }
            });
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])).join();
            log.info("表" + migrateTable + "迁移完成，耗时{}秒", (System.currentTimeMillis() - t0)/1000);
        }
        log.info("所有表迁移完成，总耗时{}秒", (System.currentTimeMillis() - startTime)/1000);
    }

}
